package com.demo.rocketmq.producer.tx;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;

import java.util.HashMap;
import java.util.Map;

/**
 * RocketMQ事务消息listener，使用事务消息必须实现一个listener。然后自动注入到producer，一个producer只有一个listener
 * 而且业务逻辑在该listener里面完成，要事先把要发送的消息定义好，然后再做业务处理才能实现事务消息，有点坑爹
 * 如果使用RocketMQTemplate，自动注入到RocketMQTemplate的producer
 * @author cyj
 **/
@Slf4j
@RocketMQTransactionListener
public class MyRocketMqLocalTransactionListener implements RocketMQLocalTransactionListener {

    private final static Map<String, RocketMQLocalTransactionState> STATE_MAP = new HashMap<>();

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        RocketMQLocalTransactionState state;
        String transId = (String)msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        int mod = 2;
        try {
            log.info("rocketmq 执行事务操作");
            if (arg instanceof Integer) {
                log.info("payload：{}", arg);
                Integer param = (Integer) arg;
                if (param % mod != 0) {
                    throw new IllegalArgumentException("payload取模后不等于0");
                }
            } else {
                throw new Exception("payload不是Integer类型");
            }
            // 设置事务状态
            state = RocketMQLocalTransactionState.COMMIT;
        } catch (IllegalArgumentException e) {
            log.error("rocketmq 执行事务操作异常", e);
            state = RocketMQLocalTransactionState.ROLLBACK;
        } catch (Exception e) {
            log.error("rocketmq 执行事务操作异常", e);
            state = RocketMQLocalTransactionState.UNKNOWN;
        }

        STATE_MAP.put(transId, state);
        // 返回事务状态给生产者
        return state;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String transId = (String)msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        log.info("回查消息 -> transId = {}, state = {} ", transId, STATE_MAP.get(transId));
        return STATE_MAP.get(transId);
    }
}
